package com.lagou.kafka.demo.interceptor;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.header.Headers;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import java.util.Map;

/**
 * @param
 * @author oyz
 * @version 1.0.4
 * @description: 发送者拦截器,该拦截器可以设置多个
 * @date 2022/8/20 13:51
 */
public class ProducerInterceptor implements org.apache.kafka.clients.producer.ProducerInterceptor {

    Logger logger = LoggerFactory.getLogger(ProducerInterceptor.class);

    @Override
    public ProducerRecord onSend(ProducerRecord record) {
        logger.info("调用了拦截器的send方法");
        // 重新封装producer
        Integer partition = record.partition();
        String topic = record.topic();
        Integer key = (Integer) record.key();
        String value = (String) record.value();
        Headers headers = record.headers();
        Long timestamp = record.timestamp();
        ProducerRecord<Integer,String> producer = new ProducerRecord<>(topic,partition,timestamp,key,value,headers);
        return producer;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        logger.info("获取拦截器返回信息");
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {
        configs.forEach((k,v)->{
            System.out.println(k+"\t"+v);
        });
    }
}
